package net.meituan;

/**
 * @description disruptor代码样例。每10ms向disruptor中插入一个元素，消费者读取数据，并打印到终端
 */

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.ThreadFactory;

/**
 * Disruptor的设计方案
 * Disruptor通过以下设计来解决队列速度慢的问题：
 * <p>
 * 环形数组结构
 * 为了避免垃圾回收，采用数组而非链表。同时，数组对处理器的缓存机制更加友好。
 * <p>
 * 元素位置定位
 * 数组长度2^n，通过位运算，加快定位的速度。下标采取递增的形式。不用担心index溢出的问题。
 * index是long类型，即使100万QPS的处理速度，也需要30万年才能用完。
 * <p>
 * 无锁设计
 * 每个生产者或者消费者线程，会先申请可以操作的元素在数组中的位置，申请到之后，直接在该位置写入或者读取数据。
 * =============================
 * 一个生产者
 * 写数据
 * 生产者单线程写数据的流程比较简单：
 * <p>
 * 申请写入m个元素；
 * 若是有m个元素可以写入，则返回最大的序列号。这儿主要判断是否会覆盖未读的元素；
 * 若是返回的正确，则生产者开始写入元素。
 * =============================
 * 多个生产者
 * 多个生产者的情况下，会遇到“如何防止多个线程重复写同一个元素”的问题。
 * Disruptor的解决方法是，每个线程获取不同的一段数组空间进行操作。
 * 这个通过CAS很容易达到。只需要在分配元素的时候，通过CAS判断一下这段空间是否已经分配出去即可。
 * <p>
 * 但是会遇到一个新问题：如何防止读取的时候，读到还未写的元素。
 * Disruptor在多个生产者的情况下，引入了一个与Ring Buffer大小相同的buffer：available Buffer。
 * 当某个位置写入成功的时候，便把availble Buffer相应的位置置位，标记为写入成功。
 * 读取的时候，会遍历available Buffer，来判断元素是否已经就绪。
 * <p>
 * 下面分读数据和写数据两种情况介绍。
 * <p>
 * 读数据
 * 生产者多线程写入的情况会复杂很多：
 * 申请读取到序号n；
 * 若writer cursor >= n，这时仍然无法确定连续可读的最大下标。
 * 从reader cursor开始读取available Buffer，一直查到第一个不可用的元素，然后返回最大连续可读元素的位置；
 * 消费者读取元素。
 * 如下图所示，读线程读到下标为2的元素，三个线程Writer1/Writer2/Writer3正在向RingBuffer相应位置写数据，
 * 写线程被分配到的最大元素下标是11。
 * <p>
 * 读线程申请读取到下标从3到11的元素，判断writer cursor>=11。
 * 然后开始读取availableBuffer，从3开始，往后读取，发现下标为7的元素没有生产成功，于是WaitFor(11)返回6。
 * <p>
 * 然后，消费者读取下标从3到6共计4个元素。
 */
public class DisruptorMain {
    public static void main(String[] args) throws Exception {
        // 队列中的元素
        class Element {
            private int value;

            public int get() {
                return value;
            }

            public void set(int value) {
                this.value = value;
            }
        }
        // RingBuffer生产工厂,初始化RingBuffer的时候使用
        EventFactory<Element> factory = () -> new Element();

        // 生产者的线程工厂
        ThreadFactory threadFactory = r -> new Thread(r, "simpleThread");

        // 处理Event的handler
        EventHandler<Element> handler = (element, sequence, endOfBatch) -> System.out.println("消费数据: " + element.get());

        // 阻塞策略
        BlockingWaitStrategy strategy = new BlockingWaitStrategy();

        // 指定RingBuffer的大小
        int bufferSize = 16;

        // 创建disruptor，采用单生产者模式
        Disruptor<Element> disruptor = new Disruptor(factory, bufferSize, threadFactory, ProducerType.SINGLE, strategy);

        // 设置EventHandler
        disruptor.handleEventsWith(handler);

        // 启动disruptor的线程
        disruptor.start();

        RingBuffer<Element> ringBuffer = disruptor.getRingBuffer();

        for (int l = 0; true; l++) {
            // 获取下一个可用位置的下标
            long sequence = ringBuffer.next();
            try {
                // 返回可用位置的元素
                Element event = ringBuffer.get(sequence);
                // 设置该位置元素的值
                event.set(l);
                System.out.println("生产数据："+event.get());
            } finally {
                ringBuffer.publish(sequence);
            }
            Thread.sleep(10);
        }
    }
}
